package com.lagou.rpc.provider.server;

import com.lagou.rpc.provider.handler.RpcServerHandler;
import com.lagou.zkutils.GetCuratorClient;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

/**
 *  netty的启动类
 */
@Service
public class RpcServer implements DisposableBean {


    @Autowired
    private RpcServerHandler rpcServerHandler;

    // netty需要bossGroup和workerGroup，在容器销毁时，在destroy()方法中关闭线程组
    private NioEventLoopGroup bossGroup;
    private NioEventLoopGroup workerGroup;

    public void startServer(String ip,int port){
        try {
            // 1. 创建线程组
            bossGroup = new NioEventLoopGroup(1);
            workerGroup = new NioEventLoopGroup();
            // 2. 创建服务端启动助手
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 3. 设置参数
            serverBootstrap.group(bossGroup,workerGroup)
                            .channel(NioServerSocketChannel.class) // 设置通道实现
                            .childHandler(new ChannelInitializer<SocketChannel>() {
                                @Override
                                protected void initChannel(SocketChannel socketChannel) throws Exception {
                                    ChannelPipeline pipeline = socketChannel.pipeline();
                                    // 添加针对  String的编解码器，因为服务提供者和消费者，是通过json字符串传递的
                                    pipeline.addLast(new StringDecoder());
                                    pipeline.addLast(new StringEncoder());
                                    // 添加业务处理类   TODO
                                    pipeline.addLast(rpcServerHandler);
                                }
                            });
            // 4. 绑定端口
            ChannelFuture channelFuture = serverBootstrap.bind(ip, port).sync();
            System.out.println("------服务端启动成功-------");

            // 注册服务端信息到zookeeper
            connectZkAndRegister(ip,port);

            // 5. 监听端口关闭的状态
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            if (bossGroup != null){
                bossGroup.shutdownGracefully();
            }

            if (workerGroup != null){
                workerGroup.shutdownGracefully();
            }
        }
    }

    /**
     *  将服务端的信息（ip + port） 注册到zookeeper上
     */
    private void connectZkAndRegister(String ip, int port) {
        CuratorFramework zkClient = GetCuratorClient.getZkClient();
        String path = GetCuratorClient.path + "/" + ip + ":" + port;
        try {
            zkClient.create().creatingParentContainersIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    //            服务器端，在zk上存储请求的响应时间，默认0s   当前请求时间 + 响应时间
                    .forPath(path,(System.currentTimeMillis()+":"+0).getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //  实现DisposableBean  重写方法，在容器关销毁时候执行该方法，在该方法中关闭线程组
    @Override
    public void destroy() throws Exception {
        if (bossGroup != null){
            bossGroup.shutdownGracefully();
        }

        if (workerGroup != null){
            workerGroup.shutdownGracefully();
        }
    }
}
